package com.atguigu.gmall.realtime.dwd.app;


import com.atguigu.gmall.realtime.common.base.BaseSqlApp;
import com.atguigu.gmall.realtime.common.constant.Constant;
import com.atguigu.gmall.realtime.common.util.FlinkSqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * 交易域取消订单事务事实表
 * <p>
 * 实现方式一:
 * 筛选 order_detail , order_detail_activity ,order_detail_coupon 这三个表的数据都是在下单过程中产生的
 * 取消订单的时候, order_info中的数据会发生改变, 从业务上来讲, 取消订单一般在下单后的15min内, 超过15min自动取消
 * <p>
 * 本次, 状态的TTL要设置为15min+5s, 考虑业务的滞后 和 传输的延迟.
 * <p>
 * <p>
 * 实现方式二:
 * 直接从 下单事务事实表, 读取数据    因为取消订单一定是在下单之后, 且只能取消下单的数据
 * 取消的订单一定在下单表中存在
 * 取消订单的时候, order_info中的数据会发生改变, 从业务上来讲, 取消订单一般在下单后的15min内, 超过15min自动取消
 * <p>
 * 本次, 状态的TTL要设置为15min+5s, 考虑业务的滞后 和 传输的延迟.
 */
public class DwdTradeOrderCancelDetail extends BaseSqlApp {
    public static void main(String[] args) {
        new DwdTradeOrderCancelDetail().start(10015, 4, Constant.TOPIC_DWD_TRADE_ORDER_CANCEL);
    }

    @Override
    public void handle(StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        // 先设置状态的TTL
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(60 * 15 + 5));

        // 读取topic_db的数据
        readOdsTopicDb(tableEnv, Constant.TOPIC_DWD_TRADE_ORDER_CANCEL);

        // 过滤取消订单数据
        Table orderCancel = tableEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `ts` " +
                " from topic_db " +
                " where `database`='gmall' " +
                " and `table`='order_info' " +
                " and `type`='update' " +
                " and `old`['order_status']='1001' " +
                " and `data`['order_status']='1003' ");
        tableEnv.createTemporaryView("order_cancel", orderCancel);


        // 读取 dwd 层下单事务事实表数据
        tableEnv.executeSql(
                " create table " + Constant.TOPIC_DWD_TRADE_ORDER_DETAIL + "(" +
                        " id string," +
                        " order_id string," +
                        " user_id string," +
                        " sku_id string," +
                        " sku_name string," +
                        " order_price string," +
                        " province_id string," +
                        " activity_id string," +
                        " activity_rule_id string," +
                        " coupon_id string," +
                        " sku_num string," +
                        " split_original_amount string," +
                        " split_activity_amount string," +
                        " split_coupon_amount string," +
                        " split_total_amount string," +
                        " ts bigint " +
                        ")" + FlinkSqlUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL, Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));

        // join
        Table result = tableEnv.sqlQuery(
                "select  " +
                        "od.id," +
                        "od.order_id," +
                        "od.user_id," +
                        "od.sku_id," +
                        " order_price," +
                        "od.sku_name," +
                        "od.province_id," +
                        "od.activity_id," +
                        "od.activity_rule_id," +
                        "od.coupon_id," +
                        "od.sku_num," +
                        "od.split_original_amount," +
                        "od.split_activity_amount," +
                        "od.split_coupon_amount," +
                        "od.split_total_amount," +
                        "oc.ts " +
                        "from dwd_trade_order_detail od " +
                        "join order_cancel oc " +
                        "on od.order_id=oc.id ");

        // 写出
        tableEnv.executeSql(
                "create table "+Constant.TOPIC_DWD_TRADE_ORDER_CANCEL+"(" +
                        " id string," +
                        " order_id string," +
                        " user_id string," +
                        " sku_id string, " +
                        " order_price string, " +
                        " sku_name string," +
                        " province_id string," +
                        " activity_id string," +
                        " activity_rule_id string," +
                        " coupon_id string," +
                        " sku_num string," +
                        " split_original_amount string," +
                        " split_activity_amount string," +
                        " split_coupon_amount string," +
                        " split_total_amount string," +
                        " ts bigint , " +
                        " PRIMARY KEY(id) NOT ENFORCED " +
                        ")" + FlinkSqlUtil.getUpsertKafkaSinkDDl(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL));

        result.executeInsert(Constant.TOPIC_DWD_TRADE_ORDER_CANCEL);
    }
}
